package com.company.wushuo.justfortest.reactive.internal.operators;

import com.company.wushuo.justfortest.reactive.OnSubscribe;
import com.company.wushuo.justfortest.reactive.Subscriber;

/**
 * Created by wushuo on 2021/4/6
 */
public class OnSubscribeLift<T, R> implements OnSubscribe<R> {

    private OnSubscribe<T> source;
    private Operator<R, T> operator;

    public OnSubscribeLift(OnSubscribe<T> parent, Operator<R, T> operator) {
        this.source = parent;
        this.operator = operator;
    }

    @Override
    public void call(final Subscriber<R> subscriber) {
        Subscriber<T> st = operator.call(subscriber);
        try {
            // new Subscriber created and being subscribed with so 'onStart' it
            st.onStart();
            source.call(st);
        } catch (Throwable e) {
            // localized capture of errors rather than it skipping all operators
            // and ending up in the try/catch of the subscribe method which then
            // prevents onErrorResumeNext and other similar approaches to error handling
            st.onError(e);
        }
    }
}
